package com.ihr360.job.core.item;

import com.ihr360.job.core.StepListener;
import com.ihr360.job.core.entity.StepContribution;
import com.ihr360.job.core.listener.MulticasterBatchListener;
import com.ihr360.job.core.repeat.RepeatCallback;
import com.ihr360.job.core.repeat.RepeatContext;
import com.ihr360.job.core.repeat.RepeatOperations;
import com.ihr360.job.core.repeat.RepeatStatus;
import com.ihr360.job.core.scope.context.SimpleStepContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

/**
 * Simple implementation of the ChunkProvider interface that does basic chunk
 * providing from an {@link ItemReader}.
 *
 * @author Dave Syer
 * @author Michael Minella
 * @see ChunkOrientedTasklet
 */
public class SimpleChunkProvider<I> implements ChunkProvider<I> {

    protected final Logger logger = LoggerFactory.getLogger(SimpleChunkProvider.class.getName());

    protected final ItemReader<? extends I> itemReader;

    private final MulticasterBatchListener<I, ?> listener = new MulticasterBatchListener<I, Object>();

    private final RepeatOperations repeatOperations;

    public SimpleChunkProvider(ItemReader<? extends I> itemReader, RepeatOperations repeatOperations) {
        this.itemReader = itemReader;
        this.repeatOperations = repeatOperations;
    }

    /**
     * Register some {@link StepListener}s with the handler. Each will get the
     * callbacks in the order specified at the correct stage.
     *
     * @param listeners
     */
    public void setListeners(List<? extends StepListener> listeners) {
        for (StepListener listener : listeners) {
            registerListener(listener);
        }
    }

    /**
     * Register a listener for callbacks at the appropriate stages in a process.
     *
     * @param listener a {@link StepListener}
     */
    public void registerListener(StepListener listener) {
        this.listener.register(listener);
    }

    /**
     * @return the listener
     */
    protected MulticasterBatchListener<I, ?> getListener() {
        return listener;
    }

    /**
     * Surrounds the read call with listener callbacks.
     * @return item
     * @throws Exception
     */
    protected final I doRead(SimpleStepContext simpleStepContext) throws Exception {
        try {
            listener.beforeRead(simpleStepContext);
            I item = itemReader.read();
            if(item != null) {
                listener.afterRead(item);
            }
            return item;
        }
        catch (Exception e) {
            if (logger.isDebugEnabled()) {
                logger.debug(e.getMessage() + " : " + e.getClass().getName());
            }
            listener.onReadError(simpleStepContext,e);
            throw e;
        }
    }

    @Override
    public Chunk<I> provide(final StepContribution contribution,SimpleStepContext simpleStepContext) throws Exception {

        final Chunk<I> inputs = new Chunk<I>();
        repeatOperations.iterate(new RepeatCallback() {

            @Override
            public RepeatStatus doInIteration(final RepeatContext context) throws Exception {
                I item = null;
                try {
                    item = read(contribution, inputs,simpleStepContext);
                }
                catch (SkipOverflowException e) {
                    // read() tells us about an excess of skips by throwing an
                    // exception
                    return RepeatStatus.FINISHED;
                }
                if (item == null) {
                    inputs.setEnd();
                    return RepeatStatus.FINISHED;
                }
                inputs.add(item);
                contribution.incrementReadCount();
                return RepeatStatus.CONTINUABLE;
            }

        });

        return inputs;

    }

    @Override
    public void postProcess(StepContribution contribution, Chunk<I> chunk) {
        // do nothing
    }

    /**
     * Delegates to {@link #doRead()}. Subclasses can add additional behaviour
     * (e.g. exception handling).
     *
     * @param contribution the current step execution contribution
     * @param chunk the current chunk
     * @return a new item for processing
     *
     * @throws SkipOverflowException if specifically the chunk is accumulating
     * too much data (e.g. skips) and it wants to force a commit.
     *
     * @throws Exception if there is a generic issue
     */
    protected I read(StepContribution contribution, Chunk<I> chunk,SimpleStepContext simpleStepContext) throws SkipOverflowException, Exception {
        return doRead(simpleStepContext);
    }

}
